package com.swak.ARQ.udp;

import java.util.Map;
import java.util.concurrent.TimeUnit;

import com.swak.Constants;
import com.swak.OS;
import com.swak.ARQ.DataMock;
import com.swak.exception.RemotingException;
import com.swak.reactivex.arq.ARQ;
import com.swak.reactivex.arq.ARQ.Segment;
import com.swak.reactivex.threads.Contexts;
import com.swak.reactivex.threads.EventLoop;
import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.ExchangeHandlerAdapter;
import com.swak.reactivex.transport.ExchangePeer;
import com.swak.reactivex.transport.Exchangers;
import com.swak.reactivex.transport.Message;
import com.swak.reactivex.transport.TransportMode;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;
import com.swak.utils.Maps;

import io.netty.buffer.ByteBuf;

public class Client extends ExchangeHandlerAdapter implements ARQ.Flusher {

	private LoopResources resources = Contexts.createEventLoopResources(TransportMode.OS, 1, 2, "Test.EventLoop-",
			false, 2, TimeUnit.SECONDS);

	ExchangePeer node;
	URL url;
	ARQ arq;
	EventLoop loop = EventLoop.newEagerEventLoop();

	public Client() throws RemotingException {
		Map<String, String> parameters = Maps.newHashMap();
		parameters.put(Constants.Bind_Host, Constants.ANYHOST);
		this.url = new URL("Test", OS.ip(), 10889, parameters);
		this.node = Exchangers.peer(url, resources, this);
		this.arq = new ARQ(1L, true, this);
		this.runFlushInTask();
		this.runFlushOutTask();
	}

	private void runFlushInTask() {
		this.loop.schedule(() -> {
			this.arq.flushIn();
			// next task
			this.runFlushInTask();
		}, this.arq.getInterval(), TimeUnit.MILLISECONDS);
	}

	private void runFlushOutTask() {
		this.loop.schedule(() -> { 
			this.arq.flushOut();
			// next task
			this.runFlushOutTask();
		}, this.arq.getRto(), TimeUnit.MILLISECONDS);
	}

	@Override
	public void input(Segment segment, ByteBuf buffer) throws RemotingException {

	}

	@Override
	public void output(Segment segment, ByteBuf buffer) throws RemotingException {
		Message message = Message.forStream().setData(buffer).address("127.0.0.1", 10890);
		this.node.send(message);
	}

	@Override
	public void received(Channel channel, Object msg) throws RemotingException {
		this.loop.execute(() -> {
			Message message = (Message) msg;
			ByteBuf buffer = (ByteBuf) message.getData();
			this.arq.recv(buffer);
			buffer.release();
		});
	}

	@Override
	public void caught(Channel channel, Throwable exception) throws RemotingException {
		super.caught(channel, exception);
		if (exception != null) {
			exception.printStackTrace();
		}
	}

	public void send(byte[] data) throws RemotingException {
		this.loop.execute(() -> {
			ByteBuf buffer = this.arq.allotByteBuf(data.length);
			buffer.writeBytes(data);
			this.arq.send(buffer);
			buffer.release();
		});
	}

	public static void main(String[] args) throws RemotingException {
		Client client = new Client();
		client.send(DataMock.createData(1024 * 10));
	}
}
